package com.lry.rpc.netty.server;

import com.lry.common.CommonConstants;
import com.lry.filter.FilterFactory;
import com.lry.protocol.RpcInvocation;
import com.lry.protocol.RpcMessage;
import com.lry.protocol.RpcResult;
import com.lry.rpc.config.RpcServiceBean;
import com.lry.rpc.netty.MsgType;
import com.lry.rpc.netty.processor.Processor;
import com.lry.rpc.netty.processor.ProcessorFactory;
import com.lry.util.CollectionUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;


@Slf4j
public class ServerHandler extends ChannelInboundHandlerAdapter {

    private void closeChannelHandlerContext(ChannelHandlerContext ctx) {
        if (log.isInfoEnabled()) {
            log.info("closeChannelHandlerContext channel:" + ctx.channel());
        }
        ctx.disconnect();
        ctx.close();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof RpcMessage) {
            RpcMessage rpcMessage = (RpcMessage) msg;
            log.info("read:" + rpcMessage.getBody());
            try{
                FilterFactory.executeAllServerFilter(rpcMessage);
                Processor processor = ProcessorFactory.getProcessor(rpcMessage.getMessageType());
                processor.process(ctx.channel(),rpcMessage);
            }catch (Exception e){
                String err = e.getMessage();
                rpcMessage.setMessageType(MsgType.RPC_RESPONSE.getCode());
                RpcResult rpcResult = new RpcResult(null,err);
                rpcMessage.setBody(rpcResult);
                ctx.channel().writeAndFlush(rpcMessage);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)   {
        log.error(ctx.channel() + " connect exception. " + cause.getMessage(), cause);
//        try {
//            closeChannelHandlerContext(ctx);
//        } catch (Exception e) {
//            log.error("failed to close channel {}: {}", ctx.channel(), e.getMessage(), e);
//        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("inactive:" + ctx);
        closeChannelHandlerContext(ctx);
        super.channelInactive(ctx);
    }


    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            log.info("idle:" + evt);
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            if (idleStateEvent.state() == IdleState.READER_IDLE) {
                if (log.isInfoEnabled()) {
                    log.info("channel:" + ctx.channel() + " read idle.");
                }
                try {
                    closeChannelHandlerContext(ctx);
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
            }
        }
    }
}
